package com.eshore.textanalyzer.elk.service.impl;

import com.eshore.textanalyzer.elk.common.ESRestClient;
import com.eshore.textanalyzer.elk.common.ResponseMessage;
import com.eshore.textanalyzer.elk.dao.TestDao;
import com.eshore.textanalyzer.elk.pojo.TagEsState;
import com.eshore.textanalyzer.elk.pojo.ToEsData;
import com.eshore.textanalyzer.elk.pojo.Paper;
import com.eshore.textanalyzer.elk.service.ElkService;
import com.eshore.textanalyzer.elk.utils.BulkProcessorUtil;
import com.eshore.textanalyzer.elk.utils.DBHepler;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.IncludeExclude;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.io.File;
import java.io.IOException;
import java.sql.*;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * @author xyd
 * @date 2021/5/17
 * 实现 elk操作接口
 */
@Slf4j
@Qualifier
@Service("ELKServiceImpl")
public class ElkServiceImpl implements ElkService {
    // 无密码登录
//    @Autowired
//    private RestHighLevelClient client;

    // 带密码配置的
    @Autowired
    private ESRestClient restClient;

    // product操作类
    private TestDao testDao;

    /**
     * 查询索引的值
     * @param page
     * @param size
     * @param texts
     * @return
     * @throws IOException
     */
    @Override
    public Paper<Map<String,String>> getEsDataList(int page, int size, String texts) throws IOException {
        // 分页参数 初始化
        if (ObjectUtils.isEmpty(page)) {
            page = 1;
        }
        if (ObjectUtils.isEmpty(size)) {
            size = 10;
        }
        // offset表示从哪里开始 size表示查多少条数据
        int offset = (page - 1) * size;
        // 获取client对象
        RestHighLevelClient client = restClient.getClient();
        // 获取查询对象
        SearchRequest request = new SearchRequest();
        // 指定查询的索引mydata
        request.indices("mydata");
        // 创建 组合条件 对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 分页
        sourceBuilder.from(offset);
        sourceBuilder.size(size);
        // 条件1、字段pInfo中包含“苹果”
        sourceBuilder.query(QueryBuilders.matchQuery("pInfo",texts));
        // 设置高亮参数
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("pInfo");
        // 是否多个字段都高亮
        highlightBuilder.requireFieldMatch(false);
        // 设置高亮显示前后缀 选填
        highlightBuilder.preTags("<span style='color: red'>"); //前缀
        highlightBuilder.postTags("</span>");
        // 将参数添加进去
        sourceBuilder.highlighter(highlightBuilder);
        System.out.println("检索参数：" + sourceBuilder.toString());

        // 执行搜索
        request.source(sourceBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        System.out.println("======response=====" + response);

        // 获取高亮显示的结果要执行两次getHits
        // 第一次调用获取查到的数据，第二次调用获取真正命中的结果
        SearchHits hits = response.getHits();
        SearchHit[] searchHits = hits.getHits();
        System.out.println("===searchHits===" + searchHits);
        // 获取查询总数
        CountRequest countRequest = new CountRequest("mydata");
        SearchSourceBuilder sourceBuilder1 = new SearchSourceBuilder();
        sourceBuilder1.query(QueryBuilders.matchAllQuery());
        // 汇总条件
        sourceBuilder1.query(QueryBuilders.matchQuery("pInfo",texts));
        countRequest.source(sourceBuilder1);
        CountResponse countResponse = client.count(countRequest,RequestOptions.DEFAULT);
        long count = countResponse.getCount();
        System.out.println("======countResponse======" + count);
        // 将高亮数据添加进来
        List<Map<String,String>> list = new ArrayList<Map<String,String>>();
        Map<String,String> map = new HashMap<>();
        for (SearchHit hit:searchHits) {
            // 获取高亮的数据
            Map<String, HighlightField> highlightFieldMap = hit.getHighlightFields();
            // 获取文档id
            System.out.println("====hit====" + hit.getId());
            Text[] fragments = highlightFieldMap.get("pInfo").fragments();
            String newTitle = "";
            for (Text text: fragments){
                System.out.println(text.toString());
                newTitle += text;
            }
            map.put("id",hit.getId());
            map.put("value",newTitle);
            list.add(map);
        }
        // 封装分页
        Paper<Map<String,String>> listPaper = new Paper<Map<String,String>>();
        listPaper.setData(list);
        listPaper.setCount(count);
        listPaper.setPage(page);
        listPaper.setSize(size);

        return listPaper;
    }

    /**
     * 根据索引 文档_id 查询文档信息
     * @param id
     * @return
     * @throws IOException
     */
    @Override
    public List<Map<String, Object>> getSearchEsById(String id) throws IOException {
        // 获取client对象
        RestHighLevelClient client = restClient.getClient();
        SearchRequest request = new SearchRequest();
        request.indices("mydata");
        // 创建 组合条件 对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.idsQuery().addIds(id));
        System.out.println("===检索参数===" + sourceBuilder);
        // 执行搜索
        request.source(sourceBuilder);
        Map<String, Object> map = new HashMap<>();
        List<Map<String, Object>> mapList = new ArrayList<>();
        SearchResponse response = client.search(request,RequestOptions.DEFAULT);
        // 获取高亮显示的结果要执行两次getHits
        // 第一次调用获取查到的数据，第二次调用获取真正命中的结果
        SearchHits hits = response.getHits();
        System.out.println("查询结果有：" + hits.getTotalHits() + "条");
        Iterator<SearchHit> iterator = hits.iterator();
        while (iterator.hasNext()) {
            SearchHit searchHit = iterator.next();
            map = searchHit.getSourceAsMap();
            mapList.add(map);
        }
        return mapList;
    }

    /**
     * 批量从mysql导入数据库
     */
    //@Async("taskExecutor") // 开启线程，由线程实现
    @Override
    public ResponseMessage writeMysqlDataToEs(ToEsData jdbcs, TagEsState tagEsState) throws SQLException {
        System.out.println("jdbcs" + jdbcs);
        // 获取client对象
        RestHighLevelClient client = restClient.getClient();
        long startTime = System.currentTimeMillis();
        BulkProcessorUtil bulkProcessorUtil = new BulkProcessorUtil();
        BulkProcessor bulkProcessor = bulkProcessorUtil.getBulkProcessor(client,tagEsState);
        // 数据库连接 后续封装
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;

        // 获取connection对象
        connection = DBHepler.getConn(jdbcs);
        log.info("=====开始导数=====");
        String sql = "select * from tag_rx_order";

        preparedStatement = connection.prepareStatement(sql,ResultSet.TYPE_FORWARD_ONLY,ResultSet.TYPE_FORWARD_ONLY);
        preparedStatement.setFetchSize(Integer.MIN_VALUE);
        resultSet = preparedStatement.executeQuery();
        ResultSetMetaData colData = resultSet.getMetaData();

        // 定义数据结构
        ArrayList<HashMap<String, String>> dataList = new ArrayList<HashMap<String, String>>();
        //bulkProcessor 添加数据的方式不多，查看官文api发现其中map键值对的方式，因此需要先将查出来的数据转换成hashmap的方式
        HashMap<String, String> map = null;
        String c = null;
        String v = null;
        int count = 0;
        while (resultSet.next()) {
            count++;
            map = new HashMap<String,String>(120);
            // 遍历将数据转为map
            for (int i = 1; i<= colData.getColumnCount(); i++) {
                c = colData.getColumnName(i);
                v = resultSet.getString(c);
                map.put(c,v);
            }
            dataList.add(map);

            // 每10w提交一次
            if (count % 100000 == 0) {
                log.info("Mysql handle data number : " + count);
                // 将数据添加到
                // 遍历集合 批量导数
                for (HashMap<String,String> hashMap:dataList) {
                    // 一次性导入数据
                    bulkProcessor.add(new IndexRequest(jdbcs.getIndexName().toLowerCase()).source(hashMap));
                }
                // 每提交一次便将map与list清空
                map.clear();
                dataList.clear();
            }
        }
        // 处理未提交的数据
        for (HashMap<String,String> hashMap:dataList) {
            bulkProcessor.add(new IndexRequest(jdbcs.getIndexName().toLowerCase()).source(hashMap));
        }
        // 将数据刷新到es 取决于bulkProcessor设置的刷新时间
        log.info("-----------------finally insert number total：" + count);
        bulkProcessor.flush();
        try {
            resultSet.close();
            preparedStatement.close();
            connection.close();
            boolean terminatedFlag = bulkProcessor.awaitClose(150L, TimeUnit.SECONDS);
            client.close();
            log.info(String.valueOf(terminatedFlag));
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
            return ResponseMessage.createByError("系统出现异常");
        }
        log.info("====数据同步时间/s====：" + (System.currentTimeMillis() - startTime)/1000 + "s");
        return ResponseMessage.createBySuccess();
    }

    /**
     * 多字段 查询
     * 对应黄埔返回数据 不需要高亮显示
     * @param page
     * @param size
     * @param texts
     * @return
     * @throws IOException
     */
    @Override
    public ResponseMessage getEsDataListHp(int page, int size, String texts) {
        // 分页参数 初始化
        if (ObjectUtils.isEmpty(page)) {
            page = 1;
        }
        if (ObjectUtils.isEmpty(size)) {
            size = 10;
        }
        // offset表示从哪里开始 size表示查多少条数据
        int offset = (page - 1) * size;
        // 获取client对象
        RestHighLevelClient client = restClient.getClient();
        // 获取查询对象
        SearchRequest request = new SearchRequest();
        // 指定查询的索引mydata
        request.indices("hpdata");
        // 创建 组合条件 对象
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 创建一个 组合查询的 多字段 and or
        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
        boolQueryBuilder.should(QueryBuilders.matchQuery("act_tag_item_values",texts));
        boolQueryBuilder.should(QueryBuilders.matchQuery("title",texts));
        // 分页
        sourceBuilder.from(offset);
        sourceBuilder.size(size);
        // 条件1、字段pInfo中包含“苹果”
//        sourceBuilder.query(QueryBuilders.matchQuery("act_tag_item_values",texts));
        sourceBuilder.query(boolQueryBuilder);
        // 设置高亮参数
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("act_tag_item_values");
        highlightBuilder.field("title");
        // 是否多个字段都高亮
        highlightBuilder.requireFieldMatch(false);
        // 设置高亮显示前后缀 选填
        highlightBuilder.preTags("<span style='color: red'>"); //前缀
        highlightBuilder.postTags("</span>");
        // 将参数添加进去
        sourceBuilder.highlighter(highlightBuilder);
        System.out.println("检索参数：" + sourceBuilder.toString());
        try {
            // 执行搜索
            request.source(sourceBuilder);
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            System.out.println("======response=====" + response);

            // 获取高亮显示的结果要执行两次getHits
            // 第一次调用获取查到的数据，第二次调用获取真正命中的结果
            SearchHits hits = response.getHits();
            SearchHit[] searchHits = hits.getHits();
            System.out.println("===searchHits===" + searchHits);
            // 获取查询总数
            CountRequest countRequest = new CountRequest("hpdata");
            SearchSourceBuilder sourceBuilder1 = new SearchSourceBuilder();
            sourceBuilder1.query(QueryBuilders.matchAllQuery());
            // 汇总条件
            sourceBuilder1.query(QueryBuilders.matchQuery("act_tag_item_values",texts));
            countRequest.source(sourceBuilder1);
            CountResponse countResponse = client.count(countRequest,RequestOptions.DEFAULT);
            long count = countResponse.getCount();
            System.out.println("======countResponse======" + count);
            // 将高亮数据添加进来
            List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
            Map<String,Object> map = new HashMap<>();
            for (SearchHit hit:searchHits) {
                // 获取高亮的数据
                Map<String, HighlightField> highlightFieldMap = hit.getHighlightFields();
                // 获取文档id
                System.out.println("====hit====" + hit.getId());
                Text[] fragments = highlightFieldMap.get("act_tag_item_values").fragments();
                String newTitle = "";
                for (Text text: fragments){
                    System.out.println(text.toString());
                    newTitle += text;
                }
                map.put("id",hit.getId());
                map.put("value",newTitle);
                list.add(map);
            }
            // 封装分页
            Paper<Map<String,Object>> listPaper = new Paper<Map<String,Object>>();
            listPaper.setData(list);
            listPaper.setCount(count);
            listPaper.setPage(page);
            listPaper.setSize(size);

            return ResponseMessage.createBySuccess(listPaper);
        } catch (IOException e) {
            e.printStackTrace();
            return ResponseMessage.createByError("数据获取失败");
        }
    }

    /**
     * 清空指定索引下的名称
     * @param indexName
     */
    @Override
    @Async("taskExecutor")
    public ResponseMessage clearEsIndexData(String indexName,String business_type,TagEsState tagEsState) {
        System.out.println("清空索引====" + indexName + "的数据");
        // 获取client对象
        RestHighLevelClient client = restClient.getClient();
        DeleteByQueryRequest request = new DeleteByQueryRequest();
        request.indices(indexName);
        request.setQuery(QueryBuilders.matchAllQuery());
        // 使用listener监听是否成功 无返回值
        ActionListener listener = new ActionListener<BulkByScrollResponse>() {
            // 监听成功响应
            @Override
            public void onResponse(BulkByScrollResponse response) {
                // 这里做日志状态更新
                log.info("deleteByQueryBigData success!,result：" + response.toString());
            }
            // 监听失败响应
            @Override
            public void onFailure(Exception e) {
                e.printStackTrace();
                log.error("deleteByQueryBigData failed,result：" + request.toString());
            }
        };
        client.deleteByQueryAsync(request, RequestOptions.DEFAULT,listener);
        return ResponseMessage.createBySuccess();
    }

    /**
     * 获取 指定月份热词分析的词频数据 并且入库
     * @param time_month
     * @return
     */
    @Override
    public ResponseMessage getHotWordsByMonth(String time_month) {
        // 初始化获取数据库的值
        ToEsData jdbcs = new ToEsData();
        jdbcs.setUrl("jdbc:mysql://192.168.115.79:3306/text_analysis?zeroDateTimeBehavior=convertToNull&useUnicode=true&characterEncoding=utf-8");
        jdbcs.setClassName("com.mysql.jdbc.Driver");
        jdbcs.setUserName("root");
        jdbcs.setPassword("root123$");
        // 数据库连接
        Connection connection = null;
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;

        // 获取connection对象
        connection = DBHepler.getConn(jdbcs);
        log.info("=====开始导数=====");
        String sql = "select * from stop_dic_config";
        ArrayList<String> dataList = new ArrayList<String>();
        try{
            preparedStatement = connection.prepareStatement(sql,ResultSet.TYPE_FORWARD_ONLY,ResultSet.TYPE_FORWARD_ONLY);
            preparedStatement.setFetchSize(Integer.MIN_VALUE);
            resultSet = preparedStatement.executeQuery();
            ResultSetMetaData colData = resultSet.getMetaData();
            String c = null;
            String v = null;
            int count = 0;
            while (resultSet.next()) {
                count++;
                // 遍历数据存入list
                for (int i = 1; i<= colData.getColumnCount(); i++) {
                    c = colData.getColumnName(i);
                    v = resultSet.getString(c);
                    dataList.add(v);
                }
            }
            // es查询
            RestHighLevelClient client = restClient.getClient();
            // 查询对象
            SearchRequest request = new SearchRequest();
            // 条件绑定
            SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
            // 查询条件
            sourceBuilder.query(QueryBuilders.termQuery("statistic_time_month_1",time_month));
            // 聚合对象
            // 聚合模板 暂时不优化先
            TermsAggregationBuilder aggr_item_values = AggregationBuilders.terms("act_tag_item_values");
            TermsAggregationBuilder aggr_title = AggregationBuilders.terms("title");
            TermsAggregationBuilder aggr_thing_subject = AggregationBuilders.terms("thing_subject");
            // 条件 -- 从三个条件中
            aggr_item_values.field("act_tag_item_values");
            aggr_item_values.size(3000);
            aggr_title.field("title");
            aggr_title.size(3000);
            aggr_thing_subject.field("thing_subject");
            aggr_thing_subject.size(3000);
            // 聚合过滤条件include只包含 -- 只匹配两个字符以上 exclude不包含 -- 清除带数字的
            IncludeExclude includeExclude = new IncludeExclude(".*[\u4E00-\u9FA5][\u4E00-\u9FA5].*",".*[(0-9)].*");
            aggr_item_values.includeExclude(includeExclude);
            aggr_title.includeExclude(includeExclude);
            aggr_thing_subject.includeExclude(includeExclude);
            // 封装条件
            sourceBuilder.aggregation(aggr_item_values);
            sourceBuilder.aggregation(aggr_title);
            sourceBuilder.aggregation(aggr_thing_subject);
            request.source(sourceBuilder);
            // 获取 响应 数据
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            System.out.println("输出一下聚合条件====" + sourceBuilder);
            // 获取聚合后的响应值
            Aggregations aggregations = response.getAggregations();
            System.out.println();
            Terms item_values = aggregations.get("act_tag_item_values");
            Terms title = aggregations.get("title");
            Terms thing_subject = aggregations.get("thing_subject");
            List<? extends Terms.Bucket> buckets_item_values = item_values.getBuckets();
            List<? extends Terms.Bucket> buckets_title = title.getBuckets();
            List<? extends Terms.Bucket> buckets_thing_subject = thing_subject.getBuckets();
            // 因为map输出是无序的，为了保证输出是有序的，因此使用链表结构LinkedHashMap
            HashMap<String, Long> map = new LinkedHashMap<>();
            // 规避一个风险 有一点点效果 就是关键词词频出现多的放在前面，避免被后续插入去重的时候覆盖
            buckets_thing_subject.forEach(el -> {
                map.put(el.getKeyAsString(),el.getDocCount());
            });
            buckets_title.forEach(el -> {
                map.put(el.getKeyAsString(),el.getDocCount());
            });
            buckets_item_values.forEach(el -> {
                map.put(el.getKeyAsString(),el.getDocCount());
            });
            // 判断map与list是否匹配 匹配则删除map （自定义的规则）
            System.out.println(map.toString());
            // 不能使用for循环删除map 会抛出异常 请使用迭代器
            Iterator<Map.Entry<String, Long>> it = map.entrySet().iterator();
            while (it.hasNext()) {
                Map.Entry<String, Long> entry = it.next();
                for (int i=0;i<dataList.size();i++) {
                    // 当map与datalist一致的时候，删除map的值 不一致则继续
                    if (dataList.get(i).equals(entry.getKey())) {
                        // 这里删除map 使用迭代器索引删除的方式
                        it.remove();
                        // 已经删除后跳出循环
                        break;
                    } else {
                        continue;
                    }
                }
            }
            // 将map数据 连同 月份 更新时间入库
            // 将map写入文本
//            String filePath = "D:/act_tag_item_values.json";
//            ObjectMapper objectMapper = new ObjectMapper();
//            objectMapper.writeValue(new File(filePath), map);
            // 关闭释放资源
            resultSet.close();
            preparedStatement.close();
            connection.close();
            // 数据封装
//            client.close();
            return ResponseMessage.createBySuccess(map);
        }catch (IOException | SQLException e) {
            e.printStackTrace();
            return ResponseMessage.createByError(e.getMessage());
        }
    }
}
